import pymysql
import warnings
import pandas as pd
from sql_insert import find_file_name
from pandas.testing import assert_frame_equal
from global_parameter import common_dir_path, ip, DB_NAME

# 忽略警告信息
warnings.filterwarnings('ignore')


def deal_data(data):
    """
    将csv和sql数据提取并去重
    :param data:
    :return:
    """
    # 初始化一个空的集合来存储去重后的数值
    unique_values = set()

    # 遍历列表中的每个字符串
    for item in data:
        # 使用逗号将字符串拆分成数值，并将其添加到集合中
        values = item.split(',')
        unique_values.update(values)

    # 将集合转换为列表
    unique_values_list = list(unique_values)
    unique_values_list = [float(i) for i in unique_values_list]

    # 断言数据列表中，数值相差不小于正负0.00000001
    if len(unique_values_list) != 1:
        # 使用断言检查条件
        # 如果断言通过，没有任何输出
        # 如果不通过，将会引发 AssertionError 异常，并显示相应的错误消息
        for i in range(1, len(unique_values_list)):
            assert abs(
                unique_values_list[i] - unique_values_list[
                    0]) < 0.00000001, f"第{i + 1}个数据与第一个数据的差值不小于0.00000001"

    return unique_values_list


def check_csv_result(dir_path, scene_name):
    """
    核对sql和csv的结果
    :param dir_path:
    :param scene_name:
    :return:
    """
    # 获取需要比较的sql数据
    con_user = pymysql.connect(host=ip, port=3306, user="oetsky", password='Oetsky@123', db=DB_NAME,
                               charset='utf8')
    sql_err_voltage_error = pd.read_sql("SELECT * FROM err_voltage_error;", con_user)
    sql_err_voltage_error_pile = pd.read_sql("SELECT * FROM err_voltage_error_pile;", con_user)
    sql_err_voltage_slide_window = pd.read_sql("SELECT * FROM err_voltage_slide_window;", con_user)
    con_user.close()

    # 获取csv文件名称
    list_file_name = find_file_name(dir_path + "\\" + scene_name)
    algo_config, da_voltage_data_temp, dv_voltage_channel, dv_voltage_line_group_channel, dv_voltage_line_group, \
        fs_voltage_transformer, err_voltage_dynamic_slide_window, err_voltage_error_pile, err_voltage_matrix_status, \
        err_voltage_pre_pile_record, err_voltage_slide_window, err_voltage_error = list_file_name

    # 读取需要比较的csv数据
    csv_err_voltage_error = pd.read_csv(err_voltage_error)
    csv_err_voltage_error_pile = pd.read_csv(err_voltage_error_pile)
    csv_err_voltage_slide_window = pd.read_csv(err_voltage_slide_window)

    # 使用drop方法删除第一列
    csv_err_voltage_error.drop(csv_err_voltage_error.columns[0], axis=1, inplace=True)
    csv_err_voltage_error_pile.drop(csv_err_voltage_error_pile.columns[0], axis=1, inplace=True)
    csv_err_voltage_slide_window.drop(csv_err_voltage_slide_window.columns[0], axis=1, inplace=True)

    # NAN使用none填充
    sql_err_voltage_error = sql_err_voltage_error.fillna("None")
    csv_err_voltage_error = csv_err_voltage_error.fillna("None")

    # 去除不需要比较的列
    sql_err_voltage_error = sql_err_voltage_error.drop(
        ["id", "create_time", "calculate_time", "data_start_time", "data_end_time"], axis=1)
    csv_err_voltage_error = csv_err_voltage_error.drop(
        ["id", "create_time", "calculate_time", "data_start_time", "data_end_time"], axis=1)

    # datafrme只取需要比较的两列
    sql_err_voltage_error_pile = sql_err_voltage_error_pile[['ratio_varargin', 'angle_varargin']]
    csv_err_voltage_error_pile = csv_err_voltage_error_pile[['ratio_varargin', 'angle_varargin']]
    sql_ratio_varargin = deal_data(sql_err_voltage_error_pile['ratio_varargin'].tolist())
    sql_angle_varargin = deal_data(sql_err_voltage_error_pile['angle_varargin'].tolist())
    csv_ratio_varargin = deal_data(csv_err_voltage_error_pile['ratio_varargin'].tolist())
    csv_angle_varargin = deal_data(csv_err_voltage_error_pile['angle_varargin'].tolist())
    print(sql_ratio_varargin)
    print(sql_angle_varargin)
    print(csv_ratio_varargin)
    print(csv_angle_varargin)
    # sql和csv取第一个值进行比较，由于sql和csv内部已经进行过断言比较，只需比较第一个值即可
    if len(csv_ratio_varargin) != 0 and len(sql_ratio_varargin) != 0:
        assert abs(
            sql_ratio_varargin[0] - csv_ratio_varargin[0]) < 0.00000001, "err_voltage_error_pile数据表ratio_varargin判断出现异常"
    if len(csv_angle_varargin) != 0 and len(sql_angle_varargin) != 0:
        assert abs(
            sql_angle_varargin[0] - csv_angle_varargin[0]) < 0.00000001, "err_voltage_error_pile数据表angle_varargin判断出现异常"

    # 去除不需要比较的列
    sql_err_voltage_slide_window = sql_err_voltage_slide_window.drop(
        ["id", "collect_time", "create_time", "pretreat_result_detail"], axis=1)
    csv_err_voltage_slide_window = csv_err_voltage_slide_window.drop(
        ["id", "collect_time", "create_time", "pretreat_result_detail"], axis=1)

    # 将整数列从 int64 转换为 object
    sql_err_voltage_error['calculate_status'] = sql_err_voltage_error['calculate_status'].astype('int64')
    csv_err_voltage_error['calculate_status'] = csv_err_voltage_error['calculate_status'].astype('int64')

    # 判断两个dataframe是否相同，不同会抛异常
    assert_frame_equal(sql_err_voltage_error, csv_err_voltage_error)
    # assert_frame_equal(sql_err_voltage_error_pile, csv_err_voltage_error_pile)
    assert_frame_equal(sql_err_voltage_slide_window, csv_err_voltage_slide_window)


if __name__ == '__main__':
    check_csv_result(r"D:\code\algorithm\算法测试-test\通用算法", "8V0【并-并，并-并-并-并-并-并】")
